package udf.matching;

import java.io.IOException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;

import model.Collocation;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;

import udf.string.CleanTag;

public class GenerateQueryTagsGeneral extends EvalFunc<DataBag> {

	BagFactory mBagFactory = BagFactory.getInstance();
	CleanTag cleaner = null;

	private boolean initialized = false;

	private Hashtable<String, String> hyphenized = new Hashtable<String, String>();
	//private Hashtable<String, String> simple = new Hashtable<String, String>();

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";

	/**
	 * Tuple fields
	 * 
	 * 0 set of tags of the bookmark 1 set of tags hpyhenized (dictionary)
	 * 
	 * 
	 * @param tuple
	 * @throws IOException
	 */

	/**
	 * Position 0 : a_b Position 1 : freq
	 * 
	 * 
	 */

	
	
	public void init(Tuple tuple) throws IOException {

		DataBag bag = (DataBag) tuple.get(4);

		Iterator<Tuple> iterator = bag.iterator();

		while (iterator.hasNext()) {

			Tuple tup = iterator.next();

			String tag = (String) tup.get(0);
			String clase="1";
			if(tup.size()>1)
			  clase = (String) tup.get(1);

			if(clase.equals("1"))
				hyphenized.put(tag, clase);
			// System.out.dfprintln("Init dictionary:" + tag + "\t"+ freq);
			// add simple
		
	

		}

		initialized = true;
	}

	
	
	/**
	 * Improvement: ADD PMI RATIO THRESHOLD TO DECIDE WHEN TO ADD BIGRAM
	 * OR WHEN TO ADD BIGRAM AND ADD SINGLE TOKENS AS WELL TO THE HASH
	 * 
	 * 
	 * 
	 * @param query
	 * @param title
	 * @param description
	 * @param threshold
	 * @return
	 */
	
	public Hashtable<String, Integer> getTags(String query, String title,
			String description, int threshold) {
		Hashtable<String, Integer> suggestions = new Hashtable<String, Integer>();

		
		//System.out.println("Tags of  query:" +query);
		LinkedList<String> temporal = splitText(query);
		//for(int i =0; i< temporal.size(); i++){
	
		//	System.out.println(temporal.get(i));
		//}
		updateHash(suggestions, temporal);

		//System.out.println("Hash frequency:");
		//printHash(suggestions);
		
		
		temporal = splitText(title);

		//System.out.println("Tags of  title:" +title);
		
		///for(int i =0; i< temporal.size(); i++){
	
		//	System.out.println(temporal.get(i));
		//}
		//System.out.println("Hash frequency:");
		
		updateHash(suggestions, temporal);
				
		//printHash(suggestions);
	
		temporal = splitText(description);

		updateHash(suggestions, temporal);

		return suggestions;
	}
	
	
	public void printHash(Hashtable<String, Integer> suggestions){
		
		Enumeration<String> keys = suggestions.keys();
		
		while(keys.hasMoreElements()){
			String key = keys.nextElement();
			System.out.println(key + "\t"+ suggestions.get(key));
			
		}
	}

	/**
	 * try to match bigrams in text
	 * 
	 * if bigram is match it is added directly
	 * 
	 * otherwise each term is added independently
	 * 
	 * @param suggestions
	 * 
	 * @param temporal
	 */

	private void updateHash(Hashtable<String, Integer> suggestions,
			LinkedList<String> temporal) {
		// TODO Auto-generated method stub


		for (int i = 0; i < temporal.size(); i++) {
			int n_gram=5;
			boolean found=false;
			while(n_gram>=0 && !found){


				if(i+n_gram<temporal.size())
				{
					
					String gram=temporal.get(i);
					int index=1;
					for(int j=i+1; j<temporal.size() && index<=n_gram; j++){

						index++;
						gram = gram + "_" + temporal.get(j);
					}
					
					
					if (hyphenized.containsKey(gram)) {

						Integer freq=1;
						if (suggestions.containsKey(gram)) {
							freq = suggestions.get(gram) + 1;

						}

						suggestions.put(gram, freq);
						found=true;
						
						i= i+n_gram;
					}

				}
				
				if(n_gram==0){
					String gram=temporal.get(i);
					Integer freq=1;
					if (suggestions.containsKey(gram)) {
						freq = suggestions.get(gram) + 1;

					}
					found=true;
					suggestions.put(gram, freq);
					
					
				}
				
				n_gram--;
				
			}
		
		}

	}

	public LinkedList<String> splitText(String text) {

		LinkedList<String> suggestions = new LinkedList<String>();

		if (text == null) {

			return suggestions;
		}

		text = text.replaceAll("\\s+", " ").toLowerCase();
		String words[] = text.split(" ");
		for (int i = 0; i < words.length; i++) {
			
			String word = CleanTag.normalizeTag(words[i]);

			
			if (!cleaner.isTagTrash(word) && !word.trim().equals("")) {
				suggestions.add(word.trim());

			}

		}

		return suggestions;

	}

	public DataBag exec(Tuple input) throws IOException {

		
		if (!initialized)
			init(input);
		
		
		if (input.get(0) == null) {
			return null;
		}

		else {

			String path = (String) input.get(3);
			String query = (String) input.get(0);
			String title = (String) input.get(1);
			String snippet = (String) input.get(2);

			if(query==null)
				query="";
			
			if(title==null)
				title="";
			if(snippet==null)
				snippet="";
			
		//	path = path.replaceAll("[.,;:_|-]", " ");
			query = query.replaceAll("[.,;:_|-]", " ");
			title = title.replaceAll("[.,;:_|-]", " ");
			snippet = snippet.replaceAll("[.,;:_|-]", " ");
			
			//System.out.println(query+"\ttitle: "+ title+ "\tsnippet:"  + snippet);
			
			int threshold = input.getType(5);
			// System.out.println(query+"\t" + title+ "\t"+ snippet);

			if (cleaner == null) {
				cleaner = new CleanTag(path);

			}

			DataBag output = mBagFactory.newDefaultBag();

			//query = query.replace("", newChar)
			
			Hashtable<String, Integer> list = getTags(query, title, snippet,
					threshold);

			if (list == null) {
				TupleFactory mTupleFactory = TupleFactory.getInstance();
				Tuple tuple = mTupleFactory.newTuple();
				tuple.append("-error-");
				tuple.append(-1);
				output.add(tuple);

				// System.out.println(query + "\t" + "-error-" + "\t" + (-1));

			} else if (list.size() == 0) {
				TupleFactory mTupleFactory = TupleFactory.getInstance();
				Tuple tuple = mTupleFactory.newTuple();
				tuple.append("-no_tags-");
				tuple.append(-2);
				output.add(tuple);

			} else {
				Enumeration<String> iter = list.keys();

				while (iter.hasMoreElements()) {
					String tag = iter.nextElement();
					TupleFactory mTupleFactory = TupleFactory.getInstance();
					Tuple tuple = mTupleFactory.newTuple();
					tuple.append(tag);
					tuple.append(list.get(tag));
					output.add(tuple);

				}

			}

			// StringTokenizer tok = new StringTokenizer((String)o,
			// " \",()*", false);
			// while (tok.hasMoreTokens())
			// output.add(mTupleFactory.newTuple(tok.nextToken()));
			// return output;

			return output;

		}

	}
	
	public static void main(String args[]){
		
		
		
		GenerateQueryTagsGeneral g = new GenerateQueryTagsGeneral();
		
		Hashtable<String,String> h = new Hashtable<String,String>();
		h.put("star_wars", "1");
		h.put("star_wars_lego_lego23_lego2_lego1", "1");
		h.put("system_of_a_down", "1");
	
		LinkedList<String> tags = new LinkedList<String>();
		tags.add("stardfa");
		tags.add("star");
		tags.add("wars");
		tags.add("lego");
		tags.add("lego23");
		tags.add("lego2");
		tags.add("lego2999");

		Hashtable<String, Integer> suggs = new Hashtable<String,Integer>();
		g.hyphenized=h;
		g.updateHash(suggs, tags);
		g.printHash(suggs);
		
		
		System.out.println("adsfadfa.;  -_=a ;.| afd a".replaceAll("[.,;:_|-]", " "));
		String t []="buenas dfasfd.adfadfa".split("[\\s+|.|,|;|:]");
		System.out.println("buenas dfasfd . adfadfa".split("[\\s.]").length);
		
		
		for(int i=0; i < t.length;i++){
			System.out.println(t[i]);
			
		}
		
	}

	public Schema outputSchema(Schema input) {
		try {
			Schema bagSchema = new Schema();

			bagSchema.add(new Schema.FieldSchema("suggestion",
					DataType.CHARARRAY));

			bagSchema.add(new Schema.FieldSchema("freq", DataType.INTEGER));

			return new Schema(new Schema.FieldSchema(getSchemaName(this
					.getClass().getName().toLowerCase(), input), bagSchema,
					DataType.BAG));

		} catch (Exception e) {
			return null;
		}
	}
}
